/*
 * Copyright 2010-2020 Alfresco Software, Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.activiti.engine.impl.jobexecutor;

import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

import org.activiti.engine.debug.Contants;
import org.activiti.engine.impl.asyncexecutor.AsyncExecutor;
import org.activiti.engine.impl.asyncexecutor.ExecuteAsyncJobCallable;
import org.activiti.engine.impl.asyncexecutor.ExecuteAsyncJobRunnable;
import org.activiti.engine.impl.cfg.ProcessEngineConfigurationImpl;
import org.activiti.engine.impl.cfg.TransactionPropagation;
import org.activiti.engine.impl.context.Context;
import org.activiti.engine.impl.interceptor.Command;
import org.activiti.engine.impl.interceptor.CommandConfig;
import org.activiti.engine.impl.interceptor.CommandContext;
import org.activiti.engine.impl.interceptor.CommandExecutor;
import org.activiti.engine.impl.persistence.entity.ExecutionJobEntityImpl;
import org.activiti.spring.SpringAsyncExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 */
public class AsyncExecutionJobNotification {

	private static Logger log = LoggerFactory.getLogger(AsyncExecutionJobNotification.class);

	protected ExecutionJobEntityImpl job;
	protected AsyncExecutor asyncExecutor;

	public AsyncExecutionJobNotification(ExecutionJobEntityImpl job, AsyncExecutor asyncExecutor) {
		this.job = job;
		this.asyncExecutor = asyncExecutor;
	}

	public void pgatewayExecute(CommandContext commandContext) {
		ProcessEngineConfigurationImpl processEngineConfiguration = commandContext.getProcessEngineConfiguration();
		CommandExecutor commandExecutor = processEngineConfiguration.getCommandExecutor();
		CommandConfig commandConfig = new CommandConfig(false, TransactionPropagation.REQUIRES_NEW);
		Map<String, Object> cacheAttr = commandContext.getAttributes();

		// 并行网关任务
		// 这里获取contextCfList，因为在下面获取，会从新生成的子线程里获取，但我们希望从外层线程获取
		List<CompletableFuture<Object>> contextCfList = Context.getPgatewayCompletableFutureList();
		boolean isParallelGatewayLastOut = job.getIsParallelGatewayLastOut();

		if(isParallelGatewayLastOut) {
			try{
				// 如果是并行网关的最后一个出度，则同步执行代码
				commandExecutor.execute(commandConfig, new Command<Void>() {
					@Override
					public Void execute(CommandContext commandContext) {
						Runnable runnable = new ExecuteAsyncJobRunnable(job, processEngineConfiguration, cacheAttr);
						runnable.run();
				        return null;
					}
				});
			}catch(Exception e){
				log.error(Contants.DEBUG_PREFIX + " last async executor error", e);
			}
		} else {
			CompletableFuture<Object> cf = commandExecutor.execute(commandConfig, new Command<CompletableFuture<Object>>() {
				@Override
				public CompletableFuture<Object> execute(CommandContext commandContext) {
					SpringAsyncExecutor springAsyncExecutor = (SpringAsyncExecutor)asyncExecutor;
					Executor executor = springAsyncExecutor.getTaskExecutor();
					return CompletableFuture.supplyAsync(() -> {
						try {
							// 这里由于是新的线程，所以需要新的上下文
							Callable<Object> callable = new ExecuteAsyncJobCallable<>(job, processEngineConfiguration, cacheAttr);
							return callable.call();
						}catch (Exception e) {
							log.error(Contants.DEBUG_PREFIX + " async executor error", e);
							return null;
						}
			        }, executor);
				}
			});
			if(cf != null){
				if(contextCfList != null){
					contextCfList.add(cf);
				}else{
					// 每次进入并行网关时(TakeOutgoingSequenceFlowsOperation)，往栈里放入新的CompletableFuture list，所以这里不可能是null
					log.error(Contants.DEBUG_PREFIX + " CompletableFuture list is null");
				}
			}
		}
	}

	public void serviceExecute(CommandContext commandContext) {
		ProcessEngineConfigurationImpl processEngineConfiguration = commandContext.getProcessEngineConfiguration();
		CommandExecutor commandExecutor = processEngineConfiguration.getCommandExecutor();
		CommandConfig commandConfig = new CommandConfig(false, TransactionPropagation.REQUIRES_NEW);
		Map<String, Object> cacheAttr = commandContext.getAttributes();

		SpringAsyncExecutor springAsyncExecutor = (SpringAsyncExecutor)asyncExecutor;
		Executor executor = springAsyncExecutor.getTaskExecutor();
		// 普通任务
		commandExecutor.execute(commandConfig, new Command<Void>() {
			@Override
			public Void execute(CommandContext commandContext) {
		        executor.execute(new ExecuteAsyncJobRunnable(job, processEngineConfiguration, cacheAttr));
		        return null;
			}
		});
	}

}
